package net.xuzhen97.distributed.transaction.order.adpter.rocketmq;

import lombok.extern.slf4j.Slf4j;
import net.xuzhen97.distributed.transaction.order.common.exception.CheckException;
import net.xuzhen97.distributed.transaction.order.service.executor.IRocketMqTransactionExe;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * rocketmq 事务监听
 */
@RocketMQTransactionListener
@Component
@Slf4j
public class TxMessageListener implements RocketMQLocalTransactionListener, ApplicationContextAware {

    private ApplicationContext applicationContext;

    /**
     * 根据业务id查找IRocketMqTransactionExe
     * @param businessId
     * @return
     */
    private IRocketMqTransactionExe getExeByBusinessId(String businessId){
        Map<String, IRocketMqTransactionExe> beansOfType = applicationContext.getBeansOfType(IRocketMqTransactionExe.class);
        List<IRocketMqTransactionExe> exes = beansOfType.values().stream()
                .filter(bean -> Objects.equals(businessId, bean.getBusinessId()))
                .collect(Collectors.toList());
        if(exes==null || exes.size() == 0){
            throw new CheckException( businessId + " IRocketMqTransactionExe不存在！！！");
        }
        if(exes.size() > 1){
            throw new CheckException( businessId + " 查找到多个IRocketMqTransactionExe， 只能有一个！！！");
        }
        return exes.get(0);
    }

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        // 消息没有业务id或者事务id则直接回滚
        String businessId = (String) message.getHeaders().get("BUSINESS_ID");
        String trxId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);

        if(businessId == null || businessId.trim().length() == 0
                || trxId == null || trxId.trim().length() == 0){
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        IRocketMqTransactionExe rocketMqTransactionExe = getExeByBusinessId(businessId);

        rocketMqTransactionExe.commit(o, trxId);

        return RocketMQLocalTransactionState.COMMIT;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        // 消息没有业务id或者事务id则直接回滚
        String businessId = (String) message.getHeaders().get("BUSINESS_ID");
        String trxId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);

        if(businessId == null || businessId.trim().length() == 0
                || trxId == null || trxId.trim().length() == 0){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        IRocketMqTransactionExe rocketMqTransactionExe = getExeByBusinessId(businessId);

        return rocketMqTransactionExe.check(trxId);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
